package com.xiezc.dht.service;

import com.xiezc.dht.message.Message;
import com.xiezc.dht.message.MessageBuild;
import com.xiezc.dht.util.Const;
import com.xiezc.dht.util.XUtil;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author wb-xzc291800
 * @date 2018/05/16 17:18
 */
@Slf4j
public class CoreService {

    @Getter
    private UdpService udpService;

    private ScheduledExecutorService scheduledExecutorService;
    private BlockingQueue<Message> sendCache;
    private BlockingQueue<Message> recvCache;
    private MessageManager messageManager;

    public CoreService(int port, BlockingQueue<Message> sendCache, BlockingQueue<Message> recvCache) {
        log.debug("开始初始化操作,udp端口:{}", port);
        this.sendCache = sendCache;
        this.recvCache = recvCache;
        udpService = new UdpService(port);
        scheduledExecutorService = new ScheduledThreadPoolExecutor(3);
        messageManager = new MessageManager(sendCache);
    }

    /**
     * 定时任务线程开始
     */
    public void timeStart() {
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            log.debug("定时任务开始 ........");
            if (Const.routerTable.refreshAndCheckEmpty()) {
                joinDHT();
            }
        }, 0, Const.deadNodeTimeSec, TimeUnit.MILLISECONDS);
    }

    /**
     * 处理线程开始
     */
    public void dealMsgStart() {
        Thread thread = XUtil
                .defaultThreadFactory
                .newThread("dealMessage", () -> {
                    while (true) {
                        try {
                            Message message = recvCache.poll(Const.cachePollSec, TimeUnit.SECONDS);
                            if (Objects.nonNull(message)) {
                               // log.error("取出一条接收到的消息开始处理");
                                messageManager.execute(message);
                            } else {
                                log.debug("dealMsgStart  超时没有拉取到数据 ........");
                            }
                        } catch (Exception e) {
                            log.error(e.getMessage(), e);
                        }
                    }
                });
        thread.start();
    }

    /**
     * 接收线程开始
     */
    public void receviceStart() {
        Thread thread = XUtil
                .defaultThreadFactory
                .newThread("receviceMessage", () -> {
                            while (true) {
                                try {
                                    Message recevice = udpService.recevice();
                                    if (Objects.isNull(recevice)) {
                                        log.error("接收到的消息解析失败");
                                    } else {
                                       // log.error("接收到消息放入revcCache, recvCache大小：{}",recvCache.size());
                                        recvCache.add(recevice);
                                    }
                                } catch (IOException e) {
                                    log.error(e.getMessage(), e);
                                }
                            }
                        }
                );
        thread.start();
    }

    /**
     * 发送线程开始
     */

    public void sendStart() {
        Thread thread = XUtil
                .defaultThreadFactory
                .newThread("sendMessage", () -> {
                    while (true) {
                        Message message = null;
                        try {
                            message = sendCache.poll(Const.cachePollSec, TimeUnit.SECONDS);
                            if (Objects.nonNull(message)) {
                               // log.error("发送的消息");
                                udpService.sendDate(message);
                            } else {
                                log.debug("sendStart  超时没有拉取到数据 ........");
                            }
                        } catch (Exception e) {
                            log.error("有问题的地址:" + message.getSocketAddress().toString());
                            log.error(e.getMessage(), e);
                        }
                    }
                });

        thread.start();
    }

    /**
     * 加入dht网络
     *
     * @throws InterruptedException
     * @throws IOException
     */
    private void joinDHT() {

        InetSocketAddress inetSocketAddress1 = new InetSocketAddress("tracker.open-internet.nl", 6969);
        InetSocketAddress inetSocketAddress2 = new InetSocketAddress("tracker.leechers-paradise.org", 6969);
        InetSocketAddress inetSocketAddress3 = new InetSocketAddress("tracker.opentrackr.org", 1337);

        Message message = MessageBuild.buildReqMsg()
                .buildATarget(Const.thisId)
                .buildAId(Const.thisId)
                .buildFindNode(inetSocketAddress1);

        Message message2 = MessageBuild.buildReqMsg()
                .buildATarget(Const.thisId)
                .buildAId(Const.thisId)
                .buildFindNode(inetSocketAddress2);

        Message message3 = MessageBuild.buildReqMsg()
                .buildATarget(Const.thisId)
                .buildAId(Const.thisId)
                .buildFindNode(inetSocketAddress3);

        for (int i = 0; i < 50; i++) {
            try {
                udpService.sendDate(message);
                udpService.sendDate(message2);
                udpService.sendDate(message3);
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
    }
}
